package org.example.service.impl;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.operators.Order;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.*;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;
import org.example.Main;
import org.example.dataset.Ceph;
import org.example.service.SearchService;
import org.example.util.TextComparator;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

@Service
public class SearchServiceImpl implements SearchService {

    @Autowired
    Ceph ceph;

    @Override
    public List inputHandle(String code) throws Exception {
        List<ByteArrayOutputStream> list = Main.list;

        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        DataSource<ByteArrayOutputStream> dataSource = env.fromCollection(list);

        //flatmap算子
        FlatMapOperator<ByteArrayOutputStream, Tuple2<String, String>> tuple2FlatMapOperator = dataSource.flatMap(new FlatMapFunction<ByteArrayOutputStream, Tuple2<String, String>>() {
            @Override
            public void flatMap(ByteArrayOutputStream byteArrayOutputStream, Collector<Tuple2<String, String>> collector) throws Exception {
                collector.collect(new Tuple2<String, String>(code, byteArrayOutputStream.toString()));
            }
        });

        //distinct去重算子
        DistinctOperator<Tuple2<String, String>> distinct = tuple2FlatMapOperator.distinct(1);

        //filter算子
        FilterOperator<Tuple2<String, String>> filter = distinct.filter(new FilterFunction<Tuple2<String, String>>() {
            @Override
            public boolean filter(Tuple2<String, String> stringTuple2) throws Exception {
                Integer abslen = Math.abs(stringTuple2.f0.length() - stringTuple2.f1.length());
                if (abslen > 0.5 * Math.max(stringTuple2.f0.length(), stringTuple2.f1.length()))
                    return false;
                else
                    return true;
            }
        });
        //sink算子
        List<Tuple2<String, String>> result = filter.collect();

        return result;
    }

    @Override
    public List<String> resultHandle(List<Tuple2<String, Double>> code_score){
        try {
            ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
            DataSource dataSource = env.fromCollection(code_score);
            //将结果按照相似度分数进行降序排序
            SortPartitionOperator sortPartitionOperator = dataSource.sortPartition(1, Order.DESCENDING);
            //取出前50个最相似的代码
            GroupReduceOperator top20 = sortPartitionOperator.first(20);
            FlatMapOperator flatMapOperator = top20.flatMap(new FlatMapFunction<Tuple2<String, Double>, String>() {
                @Override
                public void flatMap(Tuple2<String, Double> stringDoubleTuple2, Collector<String> collector) throws Exception {
                    collector.collect(stringDoubleTuple2.f0);
                }
            });
            List collect = flatMapOperator.collect();

            return collect;
        }catch (Exception e){
            return null;
        }

    }

    @Override
    public List<Tuple2<String, Double>> searchCode(List<Tuple2<String, String>> codeTuple) {
        List<Tuple2<String, Double>> Code_Score_Tuples = new ArrayList<>();
        for (Tuple2<String, String> tuple : codeTuple) {
            String code1 = tuple.f0;
            String code2 = tuple.f1;
            double similarity = TextComparator.getCosineSimilarity(code1, code2);
//            System.out.println("Cosine Similarity between documents: " + similarity);

            Code_Score_Tuples.add(new Tuple2<>(code2, similarity));

        }
        return Code_Score_Tuples;
    }


}
